import grpc
import threading
import queue
import time
import uuid
from google.protobuf.any_pb2 import Any
import signal_pb2
import signal_pb2_grpc

import argparse
import asyncio
import json
import logging
import os
import ssl
import uuid

import cv2
from aiohttp import web
from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription
from aiortc.contrib.media import MediaBlackhole, MediaPlayer, MediaRecorder, MediaRelay
from av import VideoFrame

SEND_INTERVAL = 2  # Send message every 2 seconds

send_queue = queue.Queue()
recv_queue = queue.Queue()

pc = RTCPeerConnection()


@pc.on("icecandidate")
def on_icecandidate(candidate):
    print("[WebRTC] New ICE candidate:", candidate)
    # 这里可以将 candidate 通过 gRPC 发送给对端
    # send message to gRPC server
    ice_msg = build_ice_candidate_msg(candidate)
    print("[WebRTC] Sending ICE candidate message:", ice_msg.guid)
    send_messages(ice_msg)


@pc.on("track")
def on_track(track):
    print("Track kind: %s received", track.kind)


@pc.on("connectionstatechange")
def on_connectionstatechange():
    print("Connection state is: ", pc.connectionState)
    if pc.connectionState == "failed":
        pc.close()

# ICE状态变化处理


@pc.on("iceconnectionstatechange")
async def on_iceconnectionstatechange():
    state = pc.iceConnectionState
    print("ICE连接状态变更: ", state)

# session_sid = str(uuid.uuid4())
session_sid = "26aba9da-1d60-4e2e-b80b-f47b33b186a5"
session_guuid = str(uuid.uuid4())


def message_generator(send_queue):
    while True:
        msg = send_queue.get()
        if msg is None:
            break
        yield msg


def build_login_msg():
    login = signal_pb2.Login()
    login.account_id = "user123"
    login.password = "pass123"
    login.tenant_id = "tenant456"
    login.device_id = "device789"
    login.robot_id = "robot012"
    login.app_type = "mobile"
    login.lang = "CH"
    login.device_first_connection = True
    login.device_first_connection_date = 20231001
    login.main_camera = "camera001"
    login.need_init_result = False

    any_login = Any()
    any_login.Pack(login)

    general_msg = signal_pb2.GeneralMsg()
    general_msg.id = "login"
    general_msg.type = "login"
    general_msg.sid = ""
    general_msg.seq = ""
    general_msg.guid = session_guuid
    print("[Main thread] login general_msg.guid:", general_msg.guid)
    general_msg.to = "servers"
    general_msg.src = "source_1ss"
    general_msg.dest = "destination_1sss"
    general_msg.root_guid = general_msg.guid
    general_msg.deadline = 10000
    general_msg.robot_id = "robot012"
    general_msg.params.CopyFrom(any_login)
    return general_msg


def build_start_media_msg():
    global pc
    global session_sid
    start_media = signal_pb2.StartMedia()
    start_media.sdp_offer = pc.localDescription.sdp
    start_media.urgent = 2
    start_media.destroy_media = False
    start_media.mode = "start"
    start_media.webrtc_asr = True

    any_start_media = Any()
    any_start_media.Pack(start_media)

    general_msg = signal_pb2.GeneralMsg()
    general_msg.id = "start_media"
    general_msg.type = "start_media"
    general_msg.sid = session_sid
    print("[Main thread] start media session_sid:", session_sid)
    # general_msg.sid = "c94a15f7-24d2-4256-876d-b198c5246eec"
    # print("general_msg.sid:", general_msg.sid)
    general_msg.seq = str(uuid.uuid4())
    general_msg.guid = session_guuid
    general_msg.to = "servers"
    general_msg.src = "source_1ss"
    general_msg.dest = "destination_1sss"
    general_msg.root_guid = session_guuid
    general_msg.deadline = 10000
    general_msg.robot_id = "robot012"

    general_msg.params.CopyFrom(any_start_media)

    return general_msg


def build_ice_candidate_msg(candidate):
    ice_candidate_msg = signal_pb2.IceCandidate()
    ice_candidate_msg.candidate = candidate
    ice_candidate_msg.mode = "start"
    ice_candidate_msg.tenant_id = "tenant456"
    ice_candidate_msg.feeder = ""
    ice_candidate_msg.id = ""

    any_ice_candidate = Any()
    any_ice_candidate.Pack(ice_candidate_msg)

    general_msg = signal_pb2.GeneralMsg()
    general_msg.id = "iceCandidate"
    general_msg.type = "iceCandidate"
    general_msg.sid = session_sid
    print("[Main thread] start media session_sid:", session_sid)
    # general_msg.sid = "c94a15f7-24d2-4256-876d-b198c5246eec"
    # print("general_msg.sid:", general_msg.sid)
    general_msg.seq = str(uuid.uuid4())
    general_msg.guid = session_guuid
    general_msg.to = "servers"
    general_msg.src = "source_1ss"
    general_msg.dest = "destination_1sss"
    general_msg.root_guid = session_guuid
    general_msg.deadline = 10000
    general_msg.robot_id = "robot012"

    general_msg.params.CopyFrom(any_ice_candidate)

    return general_msg


def sender(send_queue):
    while True:
        msg = build_login_msg()
        print("[Send thread] add to send queue:", msg.guid)
        send_queue.put(msg)
        time.sleep(SEND_INTERVAL)


def send_messages(msg):
    global send_queue
    print("[Send thread] Sending message:", msg.guid)
    send_queue.put(msg)


def receiver(responses, recv_queue):
    try:
        for response in responses:
            print("[Receive thread] response:", response)
            recv_queue.put(response)
    except grpc.RpcError as e:
        print("[Receive thread] gRPC error:", e)
    except Exception as e:
        print("[Receive thread] other exception:", e)


async def webrtc_offer():
    global pc
    print("Creating WebRTC offer (sendonly mode)...")

    try:
        player = MediaPlayer('sine=frequency=440', format='lavfi')
        audio_track = player.audio
        if audio_track:
            # 设置为 sendonly
            pc.addTransceiver(audio_track, direction="sendonly")
            print("[WebRTC] Audio track added (test tone, sendonly mode).")
        else:
            print("[WebRTC] No audio track found in MediaPlayer.")
    except Exception as e:
        print(f"[WebRTC] Failed to add audio track: {e}")
    offer = await pc.createOffer()
    await pc.setLocalDescription(offer)
    print("pc.localDescription.sdp, print sdp:")
    print(pc.localDescription.sdp)
    print("pc.localDescription.type, print type finished.")
    await asyncio.sleep(2)


async def set_remote_sdp(pc, sdp):
    await pc.setRemoteDescription(RTCSessionDescription(sdp=sdp, type="answer"))


def main():

    global send_queue
    global recv_queue

    is_logging = False
    channel = grpc.insecure_channel('10.15.228.3:51051')
    stub = signal_pb2_grpc.SignalMsgServiceStub(channel)

    # Starts send thread
    # t_send = threading.Thread(target=sender, args=(send_queue,), daemon=True)
    # t_send.start()

    # Starts receive thread
    responses = stub.MsgStream(message_generator(send_queue), timeout=60*5)
    t_recv = threading.Thread(target=receiver, args=(
        responses, recv_queue), daemon=True)
    t_recv.start()

    msg = build_login_msg()
    print("[Send thread] add to send queue:", msg.guid)
    send_messages(msg)

    asyncio.run(webrtc_offer())
    # ICE candidate 事件处理

    is_media_started = False

    # Processes messages in the main thread
    try:
        while True:
            try:
                msg = recv_queue.get(timeout=10)
                print("[Main thread] process response:", msg.id)
                global session_sid
                session_sid = msg.sid
                print("[Main thread] response session_sid:", session_sid)
                print("[Main thread] response msg type:", msg.type)

                if msg.type == "login":
                    print("[Main thread] login response received")
                    print("[Main thread] login response sid:", msg.sid)

                    is_logging = True

                if msg.type == "start_media":
                    print("[Main thread] start_media response received")

                    if msg.params:
                        start_media_response = signal_pb2.MediaResponse()
                        msg.params.Unpack(start_media_response)
                        print("[Main thread] start_media_response params:",
                              start_media_response)
                        print("[Main thread] start_media_response.code:",
                              start_media_response.code)
                        if start_media_response.sdp_answer:
                            print("[Main thread] sdp_answer:",
                                  start_media_response.sdp_answer)
                            # Set remote description for WebRTC (await 协程)
                            asyncio.run(set_remote_sdp(
                                pc, start_media_response.sdp_answer))
                # send start media message
                if is_logging:
                    if not is_media_started:
                        print("[Main thread] send start media message")
                        is_media_started = True
                        start_media_msg = build_start_media_msg()
                        start_media_msg.sid = session_sid
                        print("[Main thread] login sid:",
                              start_media_msg.sid)
                        send_messages(start_media_msg)
                if msg.type == "mediaReconnect":
                    print("[Main thread] mediaReconnect received")
                    start_media_msg = build_start_media_msg()
                    start_media_msg.sid = session_sid
                    print("[Main thread] mediaReconnect send start media sid:",
                          start_media_msg.sid)
                    send_messages(start_media_msg)

                if msg.type == "iceCandidate":
                    print("[Main thread] &&&SDSDiceCandidate response received")
                    pass
            except queue.Empty:
                print("[Main thread] keep wait...")
    except KeyboardInterrupt:
        print("Quit...")
        send_queue.put(None)  # Notifys send thread to exit


if __name__ == "__main__":
    main()
